package org.egomsl.mw.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.*;
import org.egomsl.mw.HadoopRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class ABaseOutputFormat extends OutputFormat<BytesWritable, HadoopRecord> {
    private static Logger log = LoggerFactory.getLogger(ABaseRecordWriter.class);

    public static String ABASE_TABLE_PATH_KEY = "abase.mapred.output.table.dir";


    public static void setOutputPath(Job job, String path) {
        job.getConfiguration().set(ABASE_TABLE_PATH_KEY, path);
    }

    @Override
    public RecordWriter<BytesWritable, HadoopRecord> getRecordWriter(TaskAttemptContext context)
            throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();
        String tablePath = conf.get(ABASE_TABLE_PATH_KEY);
        if(tablePath != null) {
            Path table = new Path(tablePath);
            Path regionPath = new Path(table, String.format("%04d", context.getJobID().getId()));
            return new ABaseRecordWriter(regionPath.getFileSystem(context.getConfiguration()), regionPath);
        }else {
            log.error("No output path specified by '"+ABASE_TABLE_PATH_KEY+"'");
        }
        return null;
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new OutputCommitter() {
            @Override
            public void setupJob(JobContext jobContext) throws IOException {

            }

            @Override
            public void setupTask(TaskAttemptContext taskContext) throws IOException {

            }

            @Override
            public boolean needsTaskCommit(TaskAttemptContext taskContext) throws IOException {
                return false;
            }

            @Override
            public void commitTask(TaskAttemptContext taskContext) throws IOException {

            }

            @Override
            public void abortTask(TaskAttemptContext taskContext) throws IOException {

            }
        };
    }
}
